package com.hudsonmq.spring.impl;


import com.hudson.codec.domain.Message;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.Duration;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.concurrent.CallBack;
import com.hudsonmq.spring.HudsonMQClientController;
import com.hudsonmq.spring.domain.produce.TopicQueueInfo;
import com.hudsonmq.spring.impl.produce.helper.SendMessageHelper;
import com.hudsonmq.spring.impl.produce.helper.SendMessageSyncHelper;
import com.hudsonmq.spring.utils.NettyUtil;
import com.hudsonmq.spring.utils.SnowflakeIdGeneratorUtil;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Component
@ConditionalOnProperty(prefix = "hudson", name = "produce-group")
public class HudsonMQTemplate {

    private static final int MAX_RETRY_NUM = 3;

    private static final Logger log = LoggerFactory.getLogger(HudsonMQTemplate.class);

    enum SendType {
        ASYNC,
        SYNC,
        ONEWAY
    }

    public void send(Message message) {
        send(message, null, SendType.ASYNC);
    }


    public void send(Message message, CallBack callBack) {
        send(message, callBack, SendType.SYNC);
    }

    public void onewaySend(Message message) {
        send(message, null, SendType.ONEWAY);
    }

    private void send(Message message, CallBack callBack, SendType sendType) {
        long messageId = SnowflakeIdGeneratorUtil.getMessageId();

        TopicQueueInfo topicQueueInfo = SendMessageHelper.sendMessage(messageId, message, callBack);
        afterSend(messageId, topicQueueInfo, message, callBack, sendType);
    }

    private boolean afterSend(long messageId, TopicQueueInfo topicQueue, Message message, CallBack callBack, SendType sendType) {
        CompletableFuture<Boolean> nettyFuture = null;
        if (sendType == SendType.ASYNC) {
            nettyFuture = new CompletableFuture<>();
            SendMessageSyncHelper.put(messageId, nettyFuture);
        } else if (sendType == SendType.SYNC) {
            nettyFuture = new CompletableFuture<>();
            SendMessageSyncHelper.put(messageId, nettyFuture);
        }

        if (nettyFuture != null) {
            try {
                return nettyFuture.get(3, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                int retryNum = 0;
                log.error("获取同步信息超时,重试{}", messageId);
                if (topicQueue != null) {
                    while (retryNum < MAX_RETRY_NUM) {
                        if (retrySend(messageId, topicQueue, message, callBack, retryNum)) {
                            log.info("消息重试成功");
                            return true;
                        }
                        retryNum++;
                    }
                }
            } catch (Exception e) {
                log.error("消息 {} 发送失败", messageId);
            } finally {
                SendMessageSyncHelper.removeFuture(messageId);
            }
        }
        return false;
    }

    private boolean retrySend(long messageId, TopicQueueInfo topicQueue, Message message, CallBack callBack, int retryNum) {
        Channel chanel = HudsonMQClientController.BROKER_CHANNEL_MAP.get(topicQueue.getBrokerName());
        message.setQueueId(topicQueue.getQueueId());

        NettyUtil.write(chanel, RemotingCommand.builder()
                .type(MessageTypeEnum.SEND_MESSAGE_REQ.getType())
                .opaque(messageId)
                .body(Message.serialize(message))
                .build()).addListener(f -> {
            if (f.isSuccess()) {
                log.debug("消息 {} 發送成功", messageId);
                if (callBack != null) {
                    SendMessageSyncHelper.put(messageId, callBack);
                }
            } else {
                log.error("消息{}:{}发送失败", messageId, new String(message.getBody(), StandardCharsets.UTF_8));
            }
        });

        CompletableFuture<Boolean> nettyFuture = new CompletableFuture<>();
        SendMessageSyncHelper.put(messageId, nettyFuture);

        try {
            return nettyFuture.get(3, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            log.error("message:{},第{}次重试失败", messageId, retryNum);
        } catch (Exception e) {
            log.error("message:{},第{}次发送失败", messageId, retryNum);
        } finally {
            SendMessageSyncHelper.removeFuture(messageId);
        }
        return false;
    }

    public void relaySend(Message message, Duration duration) {
        relaySend(message, null, duration, SendType.ASYNC);
    }

    public void relaySend(Message message, CallBack callBack, Duration duration) {
        relaySend(message, callBack, duration, SendType.ASYNC);
    }

    public void onewayRelaySend(Message message, Duration duration) {
        relaySend(message, null, duration, SendType.ONEWAY);
    }

    private void relaySend(Message message, CallBack callBack, Duration duration, SendType sendType) {

        long messageId = SnowflakeIdGeneratorUtil.getMessageId();

        TopicQueueInfo topicQueueInfo = SendMessageHelper.sendRelayMessage(messageId, message, duration);

        afterRelaySend(messageId, topicQueueInfo, message, callBack, duration, sendType);
    }

    private boolean afterRelaySend(long messageId, TopicQueueInfo topicQueue, Message message, CallBack callBack, Duration duration, SendType sendType) {

        CompletableFuture<Boolean> nettyFuture = null;

        if (sendType == SendType.ASYNC) {
            nettyFuture = new CompletableFuture<>();
            SendMessageSyncHelper.put(messageId, nettyFuture);
        } else if (sendType == SendType.SYNC) {
            nettyFuture = new CompletableFuture<>();
            SendMessageSyncHelper.put(messageId, nettyFuture);
        }

        if (nettyFuture != null) {
            nettyFuture.whenComplete((result, ex) -> {
                SendMessageSyncHelper.removeFuture(messageId);
            });

            try {
                return nettyFuture.get(3, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                int retryNum = 0;
                log.error("获取同步信息超时,重试{}", messageId);
                if (topicQueue != null) {
                    while (retryNum < MAX_RETRY_NUM) {
                        if (retrySend(messageId, topicQueue, message, callBack, retryNum)) {
                            return true;
                        }
                        retryNum++;
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException("发送失败进行重试");
            }
        }

        return false;
    }

}
